package com.bruce.demo.sequence.service;

import com.bruce.demo.sequence.config.GidServerConfig;
import com.bruce.demo.sequence.dao.SequenceDao;
import com.bruce.demo.sequence.dto.SequenceInfoDto;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;

import javax.sql.DataSource;
import java.util.AbstractQueue;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @author luotuan
 * @Description
 * @create 2020-05-15 17:40
 **/
@Service
public class SequenceServiceImpl implements SequenceService {
    @Autowired
    private GidServerConfig gidServerConfig;
    @Autowired
    private SequenceDao sequenceDao;
    @Autowired
    private DataSource dataSource;

    private ConcurrentHashMap<String, AbstractQueue<Long>> seqPool = new ConcurrentHashMap<>();


    @Override
    public Long getSequence(String sysCode, String code) {
        System.out.println("sysCode【" + sysCode + "】,code【" + code + "】开始从服务端获取gid");
        int tryTimes = gidServerConfig.getTryTimes();
        Long seq;
        String poolName = sysCode + "." + code;
        AbstractQueue<Long> queue = seqPool.get(poolName);
        for (int i = 0; i < tryTimes; i++) {
            if (queue == null) {
                SequenceInfoDto sequenceInfo = sequenceDao.queryOne(sysCode, code);
                if (sequenceInfo == null) {
                    sequenceInfo = buildInfoDto(sysCode, code);
                    sequenceDao.addSequence(sequenceInfo);
                }
                queue = new ConcurrentLinkedQueue<Long>();
                SequenceHandler sequenceHandler = new SequenceHandler(sequenceInfo, queue);
                Thread thread = new Thread(sequenceHandler, poolName + "_hander");
                thread.start();
                seqPool.put(poolName, queue);
            }
            seq = queue.poll();
            if (seq != null) {
                return seq;
            } else {
                try {
                    Thread.sleep(100L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
        throw new RuntimeException("经过了" + tryTimes + "次还是获取不到序列");
    }

    private SequenceInfoDto buildInfoDto(String sysCode, String code) {
        SequenceInfoDto sequenceInfo = new SequenceInfoDto();
        sequenceInfo.setSysCode(sysCode);
        sequenceInfo.setCode(code);
        sequenceInfo.setClientCacheSize(gidServerConfig.getClientCacheSize());
        sequenceInfo.setClientIncrementSize(gidServerConfig.getClientIncrementSize());
        sequenceInfo.setServerCacheSize(gidServerConfig.getServerCacheSize());
        sequenceInfo.setCreateTime(new Date());
        sequenceInfo.setMinValue(0);
        sequenceInfo.setState(1);
        return sequenceInfo;
    }

    class SequenceHandler implements Runnable {
        private SequenceInfoDto sequenceInfo;
        private AbstractQueue<Long> queue;

        public SequenceHandler(SequenceInfoDto sequenceInfo, AbstractQueue<Long> queue) {
            this.sequenceInfo = sequenceInfo;
            this.queue = queue;
        }

        @Override
        public void run() {
            while (true) {
                //开始进行扩容
                if (queue.size() < (sequenceInfo.getServerCacheSize() * gidServerConfig.getRefillPercent())) {
                    System.out.println("sysCode【" + sequenceInfo.getSysCode() + "】,code【" + sequenceInfo.getCode() + "】开始进行扩容");
                    ensureCapacity();

                } else {
                    try {
                        Thread.sleep(100L);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }

        private void ensureCapacity() {
            PlatformTransactionManager txManager = new DataSourceTransactionManager(dataSource);
            DefaultTransactionDefinition def = new DefaultTransactionDefinition();
            def.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
            TransactionStatus status = txManager.getTransaction(def);
            try {
                long minValue = sequenceDao.getMinValue(sequenceInfo.getSysCode(), sequenceInfo.getCode());
                long nowValue = minValue + sequenceInfo.getServerCacheSize() * sequenceInfo.getClientIncrementSize();
                sequenceDao.updateMinValue(sequenceInfo.getSysCode(), sequenceInfo.getCode(), nowValue);
                //提交事务并释放锁
                txManager.commit(status);
                //填充队列
                List<Long> capacityList = new ArrayList<>();
                while (minValue < nowValue) {
                    capacityList.add(minValue);
                    minValue += sequenceInfo.getClientIncrementSize();
                }
                queue.addAll(capacityList);
            }catch (Exception e){
                //异常，回滚事务并释放锁
                txManager.rollback(status);
            }

        }
    }
}
